package css.sword.annotation.kafka;

import java.util.Map;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;


/**
 * 
 *
 * @project 深圳OA
 * @package zhbg.oaxt.common.utils.testkafka
 * @file OAKafkaCommandsImpl.java 创建时间:2018年12月5日下午3:43:43
 * @title 标题（要求能简洁地表达出类的功能和职责）
 * @description 描述（简要描述类的职责、实现方式、使用注意事项等）
 * @copyright Copyright (c) 2018 中国软件与技术服务股份有限公司
 * @company 中国软件与技术服务股份有限公司
 * @module 模块: 模块名称
 * @author   史雪涛
 * @reviewer 审核人
 * @version 1.0
 * @history 修订历史（历次修订内容、修订人、修订时间等）
 *
 */
public class OAKafkaCommandsImpl implements OAKafkaCommands {

    /**
     * pool
     */
    @SuppressWarnings("rawtypes")
    private OAKafkaPool<KafkaProducer> pool;

    /**
     * 
     *@constructor 构造方法
     *@param pool 1
     */
    @SuppressWarnings("rawtypes")
    public OAKafkaCommandsImpl(OAKafkaPool<KafkaProducer> pool) {
        this.pool = pool;
    }

    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年7月25日上午11:34:09
     *@param kafkaProducer 1
     *@param conectionBroken 1
     *@author   史雪涛
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    @SuppressWarnings("rawtypes")
    protected void closeResource(KafkaProducer kafkaProducer, boolean conectionBroken) {
        try {
            if (conectionBroken) {
                pool.returnBrokenResource(kafkaProducer);
            } else {
                pool.returnResource(kafkaProducer);
            }
        } catch (Exception e) {
            if (kafkaProducer != null) {
                try {
                    kafkaProducer.close();
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
            }
        }
    }
    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年12月6日下午2:26:20
     *@param topics 参数
     *@param map  参数
     *@throws Exception
     * @author   曹敏
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void sendkafka(String topics, Map<String, Object> map) throws Exception {
        KafkaProducer kafkaProducer = null;
        boolean broken = false;
        try {
            kafkaProducer = pool.getResource();
            kafkaProducer.send(new ProducerRecord<String, String>(topics, JSONObject.toJSONString(map)));
        } catch (Exception e) {
            broken = handleJedisException(e);
            throw new Exception("400007:kafka运行错误", e);
        } finally {
            closeResource(kafkaProducer, broken);
        }
    }

    /**
     * 
     *@name    中文名称
     *@description 相关说明
     *@time    创建时间:2018年7月25日上午11:34:02
     *@param exception 1 
     *@return 1
     *@author   史雪涛
     *@history 修订历史（历次修订内容、修订人、修订时间等）
     */
    protected boolean handleJedisException(Exception exception) {
        return true;
    }
}